#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <sys/timerfd.h>
#include <unistd.h>
#include <poll.h>   // For poll()
#include <stdint.h> // For uint64_t
#include <sqlite3.h>
#include "MQTTClient.h"
#include "cJSON.h"
#include "shmem.h"
#include "msg_queue_peer.h"


#define CONFIG_FILE "/home/hq/work/007/data_report/config.json"
#define ADDRESS "tcp://localhost:1883"
#define CLIENTID "ExampleClientSub"
#define TOPIC "/app/data/up"
#define TOPIC2 "app/data/down"
#define QOS 1
#define TIMEOUT 10000L

volatile MQTTClient_deliveryToken deliveredtoken;
char config[4024] = "";

union val_t
{
    int b_val;   // bool类型存储空间
    int i_val;   // 整形值存储空间
    float f_val; // 浮点值存储空间
};

struct std_node
{
    int key;             // 唯一键值
    int type;            // 数据点类型
    int dev_type;        // 数据点属于哪个设备，根据网关支持的设备自行定义
    union val_t old_val; // 变化上报后需要更新旧值
    union val_t new_val; // 从共享内存取出最新数据，放到new_val中
    int ret;             // 默认为-1，采集成功后设置为0，采集失败再置-1
};

struct GlobalContext
{
    MQTTClient client;
    int std_num;
    struct std_node *std_array;
    sqlite3 *db;
};

//函数声明
void db_write_device_data(sqlite3 *db, int device_key, union val_t value, int data_type);
char* db_read_device_data(sqlite3 *db, int device_key, char *start_time, char *end_time);
int db_verify_user(sqlite3 *db, const char *username, const char *password);



/**
 * @brief 将设备节点信息序列化为 JSON 字符串
 *
 * 该函数会遍历设备节点数组，根据节点类型将节点信息添加到 JSON 对象中，
 * 最后将 JSON 对象序列化为字符串并存储在指定的缓冲区中。
 *
 * @param std 指向设备节点数组的指针
 * @param buf 用于存储序列化后 JSON 字符串的缓冲区指针
 * @param len 设备节点数组的长度
 * @param type 消息类型
 */
char *serialize_info(struct std_node *std, int len, int type,sqlite3 *db)
{
    char val_str[64];
    cJSON *root = cJSON_CreateObject();
    cJSON_AddNumberToObject(root, "type", type);
    cJSON_AddNumberToObject(root, "result", 0);
    cJSON *data = cJSON_CreateArray();
    cJSON_AddItemToObject(root, "data", data);

    for (int i = 0; i < len; i++)
    {
        
        db_write_device_data(db, std[i].key, std[i].new_val,std[i].type);

        cJSON *item = cJSON_CreateObject();
        cJSON_AddNumberToObject(item, "key", std[i].key);
        if (std[i].type == 1)
        {
            snprintf(val_str, sizeof(val_str), "%s", std[i].new_val.b_val?"true":"false");
            std[i].old_val.b_val = std[i].new_val.b_val;
        }
        else if (std[i].type == 2)
        {
            snprintf(val_str, sizeof(val_str), "%d", std[i].new_val.i_val);
            std[i].old_val.i_val = std[i].new_val.i_val;
        }
        else if (std[i].type == 3)
        {
            snprintf(val_str, sizeof(val_str), "%.2f", std[i].new_val.f_val);
            std[i].old_val.f_val = std[i].new_val.f_val;
        }
        cJSON_AddStringToObject(item, "val", val_str);
        cJSON_AddItemToArray(data, item);
    }
    char *json_str = cJSON_Print(root);
    cJSON_Delete(root); // 释放 cJSON 结构体本身
    return json_str;    // 返回新分配的字符串
}


/**
 * @brief SQLite 回调函数，用于处理 SELECT 查询的结果
 *
 * 对于 sqlite3_exec 执行的 SELECT 查询，此函数会在每获取一行数据时被调用。
 * 在此示例中，我们主要在 msgarrvd 中手动迭代 sqlite3_stmt，所以这个回调函数
 * 对于 INSERT/UPDATE/DELETE 操作可以传入 NULL。这里保留一个空实现。
 *
 * @param data 传递给回调函数的额外数据（此示例中未使用）
 * @param argc 列的数量
 * @param argv 列值的字符串数组
 * @param azColName 列名的字符串数组
 * @return int 0 表示成功，非 0 表示停止查询
 */
static int db_callback(void *data, int argc, char **argv, char **azColName) {
   // 此函数主要用于 sqlite3_exec 执行 SELECT 时的回调
   // 对于 INSERT/UPDATE/DELETE 操作，可以传入 NULL
   // 在本例的查询功能中，我们使用 sqlite3_prepare_v2 和 sqlite3_step 进行手动迭代，
   // 所以这里可以保持为空或用于调试打印。
   return 0;
}

/**
 * @brief 当 MQTT 消息成功送达至 MQTT 代理时调用的回调函数
 *
 * 此回调函数会在 MQTT 客户端确认消息已成功发送到 MQTT 代理后被触发。
 * 它会打印出已成功送达消息的令牌值，并将该令牌值存储到全局变量中，
 * 以便后续跟踪消息的送达状态。
 *
 * @param context 通用指针，通常用于传递上下文信息，比如 MQTT 客户端句柄
 * @param dt 消息交付令牌，每个发布的消息都有唯一的令牌，用于标识该消息
 */
void delivered(void *context, MQTTClient_deliveryToken dt)
{
    // 打印一条消息，显示具有指定令牌值的消息已成功送达
    printf("Message with token value %d delivery confirmed\n", dt);
    // 将当前消息的交付令牌赋值给全局变量，方便在其他地方跟踪消息的交付状态
    deliveredtoken = dt;
}

/**
 * @brief 当接收到 MQTT 消息时调用的回调函数
 *
 * 此回调函数会在 MQTT 客户端接收到来自订阅主题的消息时被触发。
 * 它可以用于处理接收到的消息，例如解析消息内容、转发消息等。
 *
 * @param context 通用指针，通常用于传递上下文信息，比如 MQTT 客户端句柄
 * @param topicName 接收到消息的主题名称
 * @param topicLen 主题名称的长度
 * @param message 指向接收到的 MQTT 消息结构体的指针
 * @return int 返回一个整数值，通常用于表示消息处理的结果，1 表示成功处理
 */
int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message)
{
    printf("Message arrived\n");
    printf("topic: %s\n", topicName);
    printf("message:\n ");

    char *buf = NULL;
    sqlite3 *db = ((struct GlobalContext *)context)->db;

    cJSON *root_down = cJSON_Parse(message->payload);
    cJSON *type = cJSON_GetObjectItem(root_down, "type");
    printf("%s:%d\n", type->string, type->valueint);

    // 从 context 中获取 client 句柄
    MQTTClient client = ((struct GlobalContext *)context)->client;
    MQTTClient_message pubmsg = MQTTClient_message_initializer; // 消息结构体
    MQTTClient_deliveryToken token;                             // 消息交付令牌

    switch (type->valueint)
    {
    case 1:
    {
        cJSON *limit = cJSON_GetObjectItem(root_down, "limit");
        if (!(strcmp(limit->valuestring, "all")))
        {
            buf = serialize_info(((struct GlobalContext *)context)->std_array, ((struct GlobalContext *)context)->std_num, 1,db);
            pubmsg.payload = buf;
            pubmsg.payloadlen = (int)strlen(buf);
            pubmsg.qos = QOS;
            pubmsg.retained = 0;
            MQTTClient_publishMessage(client, TOPIC, &pubmsg, &token);
            free(buf);
        }
        break;
    }
    case 2:
    {
        cJSON *data = cJSON_GetObjectItem(root_down, "data");
        cJSON *key = cJSON_GetObjectItem(data, "key");
        if(key->valueint>=300)
        {
            msg_queue_send("msg1", message->payload, message->payloadlen, 0);
        }
        else
        {
            msg_queue_send("msg", message->payload, message->payloadlen, 0);
        }
        break;
    }
    case 3:
    {
        int up_type; // 上报属性  //0-不上报，客户端主动采集；1-变化上报，即连续2次值不相等；2-周期上报
        int period;  // 上报周期
        cJSON *data = cJSON_GetObjectItem(root_down, "data");
        cJSON *type_target = cJSON_GetObjectItem(data, "type");
        cJSON *period_target = cJSON_GetObjectItem(data, "period");
        up_type = type_target->valueint;
        period = period_target->valueint;

        cJSON *root = cJSON_Parse(config);
        cJSON *report = cJSON_GetObjectItem(root, "report");
        cJSON_SetNumberValue(cJSON_GetObjectItem(report,"type"),up_type);
        cJSON_SetNumberValue(cJSON_GetObjectItem(report,"period"),period);

        char *json_str = cJSON_Print(root);
        cJSON_Delete(root); // 释放 cJSON 结构体本身

        /*解析点表*/
        FILE *fp;
        fp = fopen(CONFIG_FILE, "w");
        if (fp == NULL)
        {
            perror("fopen err");
            return -1;
        }
        fwrite(json_str, 1, strlen(json_str), fp);
        fclose(fp); // 关闭文件
        free(json_str); // 释放字符串内存

        break;
    }
    case 4:
    {
        cJSON *data = cJSON_GetObjectItem(root_down, "data");
        cJSON *key = cJSON_GetObjectItem(data, "key");
        cJSON *limit = cJSON_GetObjectItem(data, "limit");
        int device_key = key->valueint;
        cJSON *item = cJSON_GetArrayItem(limit, 0);
        char *start_time = item->valuestring;
        item = cJSON_GetArrayItem(limit, 1);
        char *end_time = item->valuestring;

        char *buf = db_read_device_data(db, device_key, start_time, end_time);
        pubmsg.payload = buf;
        pubmsg.payloadlen = (int)strlen(buf);
        pubmsg.qos = QOS;
        pubmsg.retained = 0;
        MQTTClient_publishMessage(client, TOPIC, &pubmsg, &token);
        free(buf);


        break;
    }
    case 5:
    {
        cJSON *data = cJSON_GetObjectItem(root_down, "data");
        cJSON *number = cJSON_GetObjectItem(data, "number");
        cJSON *password = cJSON_GetObjectItem(data, "password");

        int ret = db_verify_user(db, number->valuestring, password->valuestring);

        cJSON *root = cJSON_CreateObject();
        cJSON_AddNumberToObject(root, "type", 5);
        cJSON_AddStringToObject(root, "data", ret == 1 ? "true" : "false");
        buf = cJSON_Print(root);
        cJSON_Delete(root); // 释放 cJSON 结构体本身

        pubmsg.payload = buf;
        pubmsg.payloadlen = (int)strlen(buf);
        pubmsg.qos = QOS;
        pubmsg.retained = 0;
        MQTTClient_publishMessage(client, TOPIC, &pubmsg, &token);
        free(buf);
        break;
    }

    default:
        break;
    }

    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);
    cJSON_Delete(root_down);
    return 1;
}

/**
 * @brief 当 MQTT 连接断开时调用的回调函数
 *
 * 此回调函数会在 MQTT 客户端与 MQTT 代理断开连接时被触发。
 * 它可以用于处理连接断开的情况，例如重新连接、打印断开原因等。
 *
 * @param context 通用指针，通常用于传递上下文信息，比如 MQTT 客户端句柄
 * @param cause 断开连接的原因描述字符串
 */
void connlost(void *context, char *cause)
{
    printf("\nConnection lost\n");
    printf("     cause: %s\n", cause);
}

/**
 * @brief 将设备数据 (key, val, type, timestamp) 插入到 SQLite 数据库的 device_data_history 表中。
 *
 * 此函数用于封装设备数据写入操作，并自动获取当前时间戳。
 * 它使用参数绑定来防止 SQL 注入。
 *
 * @param db SQLite 数据库连接指针。
 * @param device_key 设备点位的唯一键值。
 * @param value 设备点位的当前值（使用 union val_t）。
 * @param data_type 设备点位的数据类型 (1: bool, 2: int, 3: float)。
 */
void db_write_device_data(sqlite3 *db, int device_key, union val_t value, int data_type) {
    char timestamp_str[80];
    time_t rawtime;
    struct tm *info;
    char data_value_str[64]; // 用于存储转换后的值字符串

    time(&rawtime);
    info = localtime(&rawtime);
    strftime(timestamp_str, sizeof(timestamp_str), "%Y-%m-%d %H:%M:%S", info);

    // 根据数据类型将值转换为字符串
    if (data_type == 1) { // bool
        snprintf(data_value_str, sizeof(data_value_str), "%s", value.b_val?"true":"false");
    } else if (data_type == 2) { // int
        snprintf(data_value_str, sizeof(data_value_str), "%d", value.i_val);
    } else if (data_type == 3) { // float
        snprintf(data_value_str, sizeof(data_value_str), "%.2f", value.f_val);
    } else {
        fprintf(stderr, "Unsupported data type for key %d: %d\n", device_key, data_type);
        return;
    }

    const char *insert_sql = "INSERT INTO device_data_history (timestamp, key, val, type) VALUES (?, ?, ?, ?);";
    sqlite3_stmt *stmt;
    int rc = sqlite3_prepare_v2(db, insert_sql, -1, &stmt, NULL);

    if (rc != SQLITE_OK) {
        fprintf(stderr, "Failed to prepare db_write_device_data statement: %s\n", sqlite3_errmsg(db));
        return;
    }

    sqlite3_bind_text(stmt, 1, timestamp_str, -1, SQLITE_TRANSIENT);
    sqlite3_bind_int(stmt, 2, device_key);
    sqlite3_bind_text(stmt, 3, data_value_str, -1, SQLITE_TRANSIENT);
    sqlite3_bind_int(stmt, 4, data_type);

    rc = sqlite3_step(stmt);
    if (rc != SQLITE_DONE) {
        fprintf(stderr, "db_write_device_data failed: %s\n", sqlite3_errmsg(db));
    } else {
        printf("Device data record inserted successfully (Key: %d, Val: %s, Type: %d).\n", device_key, data_value_str, data_type);
    }

    sqlite3_finalize(stmt);
}

/**
 * @brief 根据 key 值和时间范围从 device_data_history 表中读取数据。
 *
 * 此函数执行 SELECT 查询，并返回结果的 cJSON 数组。
 * 调用者需要负责释放返回的 cJSON 对象。
 *
 * @param db SQLite 数据库连接指针。
 * @param device_key 要查询的设备点位键值
 * @param start_time 查询的起始时间字符串 (YYYY-MM-DD HH:MM:SS)。
 * @param end_time 查询的结束时间字符串 (YYYY-MM-DD HH:MM:SS)。
 * @return cJSON* 包含查询结果的 cJSON 数组。如果查询失败或无结果，返回空的 cJSON 数组。
 */
char* db_read_device_data(sqlite3 *db, int device_key, char *start_time, char *end_time) 
{
    char *json_string = NULL;
    cJSON *root_history = cJSON_CreateObject();
    cJSON_AddNumberToObject(root_history, "type", 4);
    cJSON_AddNumberToObject(root_history, "result", 0);
    cJSON_AddNumberToObject(root_history, "key", device_key);
    cJSON *data_history = cJSON_CreateArray();
    cJSON_AddItemToObject(root_history, "data", data_history);


    sqlite3_stmt *stmt;
//     SELECT key, val: 指定要从数据库表中查询两列：key 和 val。
//     FROM device_data_history: 指定从 device_data_history 这张表查询。
//     WHERE key = ? AND timestamp BETWEEN ? AND ?: 这是查询的过滤条件。
    const char *query_sql = "SELECT timestamp, val FROM device_data_history WHERE key = ? AND timestamp BETWEEN ? AND ? ORDER BY timestamp ASC;";

    int rc_prep = sqlite3_prepare_v2(db, query_sql, -1, &stmt, NULL);
    if (rc_prep != SQLITE_OK) {
        fprintf(stderr, "Failed to prepare db_read_device_data statement: %s\n", sqlite3_errmsg(db));
        cJSON_SetNumberValue(cJSON_GetObjectItem(root_history, "result"), 1); // 失败返回1
        json_string = cJSON_Print(root_history);
        cJSON_Delete(root_history);
        return json_string;
    }

    // 绑定参数
    sqlite3_bind_int(stmt, 1, device_key);
    sqlite3_bind_text(stmt, 2, start_time, -1, SQLITE_TRANSIENT);
    sqlite3_bind_text(stmt, 3, end_time, -1, SQLITE_TRANSIENT);

    // 迭代查询结果并填充 'data' 数组
    while (sqlite3_step(stmt) == SQLITE_ROW) {
        cJSON *item = cJSON_CreateObject();
        // 根据 SELECT 语句的列顺序，0 是 device_key，1 是 data_value_str
        cJSON_AddStringToObject(item, "time", sqlite3_column_text(stmt, 0));
        cJSON_AddStringToObject(item, "val", (const char*)sqlite3_column_text(stmt, 1));
        cJSON_AddItemToArray(data_history, item);
    }

    // 检查是否有错误发生，不仅仅是 SQLITE_DONE
    // 如果 sqlite3_step 返回的不是 SQLITE_DONE，则表示有错误发生
    if (sqlite3_step(stmt) != SQLITE_DONE && sqlite3_step(stmt) != SQLITE_ROW) {
        fprintf(stderr, "Error during query execution: %s\n", sqlite3_errmsg(db));
        cJSON_SetNumberValue(cJSON_GetObjectItem(root_history, "result"), 1); // 失败返回1
    }

    sqlite3_finalize(stmt); // 释放语句资源

    json_string = cJSON_Print(root_history); // 将 cJSON 对象转换为 JSON 字符串
    cJSON_Delete(root_history); // 释放 cJSON 对象结构
    return json_string; // 返回新分配的 JSON 字符串，调用者需负责 free
}

/**
 * @brief 向 SQLite 数据库的 users 表中添加用户，如果用户已存在则忽略
 * 
 * 此函数使用参数绑定的方式向 users 表中插入新用户，避免 SQL 注入风险。
 * 如果用户已存在，使用 INSERT OR IGNORE 语句会忽略插入操作。
 * 
 * @param db SQLite 数据库连接指针
 * @param username 要添加的用户的用户名
 * @param password 要添加的用户的密码
 * @return int 成功返回 SQLITE_OK，失败返回 SQLite 错误码
 */
int db_add_user(sqlite3 *db, const char *username, const char *password) {
    sqlite3_stmt *stmt;
    int rc;
    const char *sql = "INSERT OR IGNORE INTO users (username, password) VALUES (?, ?);"; // 使用 ? 作为占位符

    rc = sqlite3_prepare_v2(db, sql, -1, &stmt, NULL); // 准备语句
    if (rc != SQLITE_OK) {
        fprintf(stderr, "SQL 准备失败: %s\n", sqlite3_errmsg(db));
        return rc;
    }

    sqlite3_bind_text(stmt, 1, username, -1, SQLITE_STATIC); // 绑定第一个参数 (username)
    sqlite3_bind_text(stmt, 2, password, -1, SQLITE_STATIC); // 绑定第二个参数 (password)

    rc = sqlite3_step(stmt); // 执行语句
    if (rc != SQLITE_DONE && rc != SQLITE_OK) { // SQLITE_DONE 表示语句执行完毕，没有更多行
        fprintf(stderr, "SQL 执行失败 (db_add_user_secure): %s\n", sqlite3_errmsg(db));
        sqlite3_finalize(stmt);
        return rc;
    }

    // 检查是否有行被插入
    if (sqlite3_changes(db) > 0) {
        fprintf(stdout, "用户 '%s' 添加成功 (安全方式)。\n", username);
    } else {
        fprintf(stdout, "用户 '%s' 已存在，跳过添加 (安全方式)。\n", username);
    }

    sqlite3_finalize(stmt); // 释放语句资源
    return SQLITE_OK;
}

/**
 * @brief SQLite 查询的回调函数，用于验证用户账号密码是否匹配
 * 
 * 此回调函数会在 `sqlite3_exec` 执行 SELECT 查询并获取到一行数据时被调用。
 * 若查询结果存在数据行，意味着找到了匹配的用户名和密码，将认证标志置为 1。
 * 
 * @param data 传递给回调函数的额外数据，这里是一个指向 `int` 类型的指针，用于存储认证结果
 * @param argc 结果集中列的数量
 * @param argv 结果集中每列的值组成的字符串数组
 * @param azColName 结果集中每列的名称组成的字符串数组
 * @return int 0 表示继续处理下一行数据（尽管此查询通常只返回一行）
 */
static int verify_callback(void *data, int argc, char **argv, char **azColName) {
    int *is_authenticated = (int*)data;
    if (argc > 0) {
        // 如果回调函数被调用，说明找到了匹配的用户名和密码
        // 因为我们只查询一条记录，只要进入回调就表示成功
        *is_authenticated = 1; // 设置认证成功标志
    }
    return 0; // 返回0表示继续处理下一行（尽管这里只有一行）
}

/**
 * @brief 验证用户账号和密码是否匹配
 * 
 * 该函数会检查传入的数据库句柄、用户名和密码是否有效，
 * 然后构建 SQL 查询语句，从 `users` 表中查找匹配的记录。
 * 使用回调函数处理查询结果，最终返回验证结果。
 * 
 * @param db SQLite 数据库连接指针
 * @param username 要验证的用户名
 * @param password 要验证的密码
 * @return int 返回验证结果，1 表示验证成功，0 表示验证失败，
 *         -1 表示参数无效，-2 表示数据库操作失败
 */
int db_verify_user(sqlite3 *db, const char *username, const char *password) {
    if (!db || !username || !password || strlen(username) == 0 || strlen(password) == 0) {
        fprintf(stderr, "错误: db_verify_user 参数无效 (数据库句柄、用户名或密码为空/空字符串)。\n");
        return -1; // 返回一个自定义错误码表示参数无效
    }

    char *zErrMsg = 0;
    int rc;
    char sql[256];
    int is_authenticated = 0; // 初始设为未认证

    // 构建查询语句：查找 username 和 password 都匹配的记录
    // 同样，直接拼接存在SQL注入风险，生产环境请使用参数绑定
    snprintf(sql, sizeof(sql), "SELECT 1 FROM users WHERE username = '%s' AND password = '%s' LIMIT 1;", username, password);

    // 执行查询，并将 is_authenticated 的地址作为用户数据传递给回调函数
    rc = sqlite3_exec(db, sql, verify_callback, &is_authenticated, &zErrMsg);

    if (rc != SQLITE_OK) {
        fprintf(stderr, "SQL 错误 (db_verify_user): %s\n", zErrMsg);
        sqlite3_free(zErrMsg);
        return -2; // 返回一个自定义错误码表示数据库操作失败
    } else {
        // is_authenticated 的值会在回调函数中被修改
        if (is_authenticated) {
            fprintf(stdout, "用户 '%s' 验证成功。\n", username);
        } else {
            fprintf(stdout, "用户 '%s' 验证失败 (账号或密码不匹配)。\n", username);
        }
        return is_authenticated; // 返回 1 (成功) 或 0 (失败)
    }
}

/**
 * @brief 初始化 SQLite 数据库，包括打开数据库文件并创建必要的表。
 *
 * 此函数将创建 `history.db` 数据库文件（如果不存在），并确保其中包含
 * `history_data` 表（用于存储原始 MQTT 消息日志）和
 * `device_data_history` 表（用于存储解析后的设备数据）。
 *
 * @param db_ptr 指向 sqlite3* 变量的指针，用于存储数据库连接句柄。
 * @return int 成功返回 SQLITE_OK，失败返回其他 SQLite 错误码。
 */
int db_init(sqlite3 **db_ptr) {
    char *zErrMsg = 0;
    int rc_db;

    // 打开数据库
    rc_db = sqlite3_open("history.db", db_ptr);
    if (rc_db) {
        fprintf(stderr, "无法打开数据库: %s\n", sqlite3_errmsg(*db_ptr));
        return rc_db;
    } else {
        fprintf(stdout, "数据库 history.db 打开成功\n");
    }

    // --- 创建 users 表 (包含账号和密码) ---
    const char *sql_create_users_table = "CREATE TABLE IF NOT EXISTS users("
                                         "id INTEGER PRIMARY KEY AUTOINCREMENT,"
                                         "username TEXT NOT NULL UNIQUE," // 账号，必须唯一
                                         "password TEXT NOT NULL);";     // 密码

    rc_db = sqlite3_exec(*db_ptr, sql_create_users_table, NULL, 0, &zErrMsg);
    if (rc_db != SQLITE_OK) {
        fprintf(stderr, "SQL 错误 (创建 users 表): %s\n", zErrMsg);
        sqlite3_free(zErrMsg);
        sqlite3_close(*db_ptr);
        return rc_db;
    } else {
        fprintf(stdout, "表 'users' 创建或已存在\n");
    }
    // --- 插入初始账号和密码 ---
    // 使用 INSERT OR IGNORE 确保只在用户不存在时才插入
    // 这样做可以避免每次运行 db_init 都重复插入相同的初始用户
    const char *sql_insert_initial_user = "INSERT OR IGNORE INTO users (username, password) VALUES ('admin', 'password123');";

    rc_db = sqlite3_exec(*db_ptr, sql_insert_initial_user, NULL, 0, &zErrMsg);
    if (rc_db != SQLITE_OK) {
        fprintf(stderr, "SQL 错误 (插入初始用户): %s\n", zErrMsg);
        sqlite3_free(zErrMsg);
        sqlite3_close(*db_ptr);
        return rc_db;
    } else {
        // 检查是否有行被插入。sqlite3_changes() 返回最近一次操作影响的行数。
        if (sqlite3_changes(*db_ptr) > 0) {
            fprintf(stdout, "初始用户 'admin' 插入成功。\n");
        } else {
            fprintf(stdout, "初始用户 'admin' 已存在，跳过插入。\n");
        }
    }


    // 创建 device_data_history 表 (用于存储解析后的设备数据)
    // key: 设备点位的唯一键值
    // val: 将布尔、整数、浮点值统一存储为字符串
    // type: 存储原始数据类型 (1: bool, 2: int, 3: float)，便于查询时理解
    const char *sql_create_device_data_table = "CREATE TABLE IF NOT EXISTS device_data_history("
                                               "ID INTEGER PRIMARY KEY AUTOINCREMENT,"
                                               "timestamp TEXT NOT NULL,"
                                               "key INTEGER NOT NULL,"
                                               "val TEXT NOT NULL,"
                                               "type INTEGER NOT NULL);";

    rc_db = sqlite3_exec(*db_ptr, sql_create_device_data_table, db_callback, 0, &zErrMsg);
    if (rc_db != SQLITE_OK) {
        fprintf(stderr, "SQL 错误 (创建 device_data_history 表): %s\n", zErrMsg);
        sqlite3_free(zErrMsg);
        sqlite3_close(*db_ptr);
        return rc_db;
    } else {
        fprintf(stdout, "表 'device_data_history' 创建或已存在\n");
    }

    // 可选：为 timestamp 和 device_key 添加索引以提高查询性能
    const char *sql_create_index_timestamp = "CREATE INDEX IF NOT EXISTS idx_device_data_timestamp ON device_data_history (timestamp);";
    const char *sql_create_index_key_timestamp = "CREATE INDEX IF NOT EXISTS idx_device_data_key_timestamp ON device_data_history (key, timestamp);";

    rc_db = sqlite3_exec(*db_ptr, sql_create_index_timestamp, db_callback, 0, &zErrMsg);
    if (rc_db != SQLITE_OK) {
        fprintf(stderr, "SQL 错误 (创建 timestamp 索引): %s\n", zErrMsg);
        sqlite3_free(zErrMsg);
        // 不影响主流程，可以继续
    }

    rc_db = sqlite3_exec(*db_ptr, sql_create_index_key_timestamp, db_callback, 0, &zErrMsg);
    if (rc_db != SQLITE_OK) {
        fprintf(stderr, "SQL 错误 (创建 key_timestamp 索引): %s\n", zErrMsg);
        sqlite3_free(zErrMsg);
        // 不影响主流程，可以继续
    }

    return SQLITE_OK;
}

int main(int argc, char *argv[])
{

    /*解析点表*/
    FILE *fp;
    
    int std_num; // 设备数量
    int up_type; // 上报属性  //0-不上报，客户端主动采集；1-变化上报，即连续2次值不相等；2-周期上报
    int period;  // 上报周期

    // --- 数据库初始化开始 ---
    sqlite3 *db; // SQLite 数据库连接对象
        if (db_init(&db) != SQLITE_OK) {
        return EXIT_FAILURE; // 数据库初始化失败，程序退出
    }
    // --- 数据库初始化结束 ---

    fp = fopen(CONFIG_FILE, "r");
    if (fp == NULL)
    {
        perror("fopen err");
        sqlite3_close(db); // 关闭数据库
        return -1;
    }
    // 确保字符串以null终止，并处理实际读取的字节数 !!!
    size_t bytes_read = fread(config, 1, sizeof(config) - 1, fp); // 留一个字节给 null 终止符
    if (bytes_read == 0 && !feof(fp))
    { // 如果读取到0字节且不是文件末尾，说明有读取错误
        perror("Error reading config.json");
        fclose(fp); // 及时关闭文件
        sqlite3_close(db); // 关闭数据库
        return -1;
    }
    config[bytes_read] = '\0'; // 确保 config 字符串以 null 终止
    fclose(fp); // 关闭文件


    cJSON *root = cJSON_Parse(config);
    if (!root) {
        fprintf(stderr, "Error parsing config JSON.\n");
        sqlite3_close(db); // 解析配置文件失败，关闭数据库
        return -1;
    }

    cJSON *report = cJSON_GetObjectItem(root, "report");
    up_type = cJSON_GetObjectItem(report, "type")->valueint;
    period = cJSON_GetObjectItem(report, "period")->valueint;

    cJSON *stm32 = cJSON_GetObjectItem(root, "stm32");
    cJSON *data_32 = cJSON_GetObjectItem(stm32, "data");
    int num_32 = cJSON_GetArraySize(data_32);

    cJSON *modbus = cJSON_GetObjectItem(root, "modbus");
    cJSON *data_modbus = cJSON_GetObjectItem(modbus, "data");
    int num_modbus = cJSON_GetArraySize(data_modbus);
    std_num = num_32 + num_modbus;

    /*创建共享内存*/
    size_t shm_size = sizeof(int) + sizeof(struct std_node) * std_num;
    struct shm_param para;
    if (shm_init(&para, "shm", shm_size) < 0)
    {
        printf("shm init err\n");
        cJSON_Delete(root);
        sqlite3_close(db);
        return -1;
    }
    // 获取共享内存的起始地址N
    void *shm_base_addr = shm_getaddr(&para);
    if (shm_base_addr == (void *)-1 || shm_base_addr == NULL)
    {
        printf("shm get addr err\n");
        shm_del(&para);
        cJSON_Delete(root);
        sqlite3_close(db);
        return -1;
    }

    // 将共享内存的起始部分强制转换为我们的头部,并将设备数量写入头部
    int *header = (int *)shm_base_addr;
    header[0] = std_num;

    // 数据数组紧随头部之后
    struct std_node *std_array = (struct std_node *)((char *)shm_base_addr + sizeof(int));
    union val_t val_init = {.b_val = 0, .f_val = 0.0, .i_val = 0};
    for (int i = 0; i < num_32; i++)
    {
        cJSON *item = cJSON_GetArrayItem(data_32, i);
        std_array[i].key = cJSON_GetObjectItem(item, "key")->valueint;
        std_array[i].type = cJSON_GetObjectItem(item, "type")->valueint;
        std_array[i].new_val = val_init;
        std_array[i].old_val = val_init;
        std_array[i].ret = -1;
    }

    for (int i = num_32; i < std_num; i++)
    {
        cJSON *item = cJSON_GetArrayItem(data_modbus, i-num_32);
        std_array[i].key = cJSON_GetObjectItem(item, "key")->valueint;
        std_array[i].type = cJSON_GetObjectItem(item, "type")->valueint;
        std_array[i].new_val = val_init;
        std_array[i].old_val = val_init;
        std_array[i].ret = -1;
    }

    MQTTClient client;                                                           // 创建客户端句柄
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer; // 链接选项
    MQTTClient_message pubmsg = MQTTClient_message_initializer;                  // 消息结构体
    MQTTClient_deliveryToken token;                                              // 消息交付令牌

    int rc;  // 返回码
   
    MQTTClient_create(&client, ADDRESS, CLIENTID,
                      MQTTCLIENT_PERSISTENCE_NONE, NULL); // 创建客户端实例，并指定代理地址、客户端ID等
    // 设置连接选项
    conn_opts.keepAliveInterval = 20; // 心跳间隔，20秒
    conn_opts.cleansession = 1;       // 清理会话，表示不保留离线消息和订阅
    struct GlobalContext context = {.client = client, .std_num = std_num, .std_array = std_array,.db = db};

    MQTTClient_setCallbacks(client, (void *)&context, connlost, msgarrvd, delivered); // 回调函数

    // 尝试与 MQTT 代理建立连接。
    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to connect, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }
    MQTTClient_subscribe(client, TOPIC2, QOS); // 订阅主题

    int timer_fd;
    struct itimerspec new_timer;
    int count = 0;

    // 创建一个 CLOCK_MONOTONIC 类型的定时器文件描述符
    // TFD_CLOEXEC: 确保在子进程中不会继承此FD
    timer_fd = timerfd_create(CLOCK_MONOTONIC, TFD_CLOEXEC);
    if (timer_fd == -1)
    {
        perror("timerfd_create failed");
        return 1;
    }
    printf("定时器文件描述符创建成功：fd = %d\n", timer_fd);

    // 设置定时器参数
    // 第一次到期时间
    new_timer.it_value.tv_sec = period;
    new_timer.it_value.tv_nsec = 0;
    // 周期性到期间隔
    new_timer.it_interval.tv_sec = period;
    new_timer.it_interval.tv_nsec = 0;

    // 启动定时器
    // 第一个参数是定时器FD
    // 第二个参数为0表示相对定时器 (相对于当前时间)
    // 第三个参数是新的定时器设置
    // 第四个参数为NULL表示不获取旧的定时器设置
    if (timerfd_settime(timer_fd, 0, &new_timer, NULL) == -1)
    {
        perror("timerfd_settime failed");
        close(timer_fd);
        return 1;
    }
    printf("定时器设置成功，将每%d秒触发一次。\n", period);

    // 使用 poll 监听定时器事件
    struct pollfd pfd;
    pfd.fd = timer_fd;   // 监听定时器文件描述符
    pfd.events = POLLIN; // 关注可读事件 (当定时器到期时可读)

    printf("进入主循环，等待定时器事件...\n");

    char *json_payload;

    switch (up_type)
    {
    case 0: // 刷新上报
    {
        while (1)
        {
            sleep(1);
        }
        break;
    }
    case 1: // 变化上报
    {
        while (1)
        {
            for (int i = 0; i < std_num; i++)
            {
                if (std_array[i].new_val.b_val != std_array[i].old_val.b_val ||
                    std_array[i].new_val.i_val != std_array[i].old_val.i_val ||
                    std_array[i].new_val.f_val != std_array[i].old_val.f_val)
                {
                    // 读取共享内存数据并序列化
                    json_payload = serialize_info(std_array, std_num, 1,db);
                    pubmsg.payload = json_payload;
                    pubmsg.payloadlen = (int)strlen(json_payload);
                    pubmsg.qos = QOS;
                    pubmsg.retained = 0;
                    MQTTClient_publishMessage(client, TOPIC, &pubmsg, &token); // 发布消息到指定主题
                    free(json_payload);
                    break;
                }
            }
            sleep(1);
        }
        break;
    }
    case 2: // 周期上报
    {
        while (1)
        {
            // poll 等待事件发生，-1 表示无限等待
            int ret = poll(&pfd, 1, -1);
            if (ret == -1)
            {
                perror("poll failed");
                break;
            }

            // 检查定时器FD是否可读
            if (pfd.revents & POLLIN)
            {
                uint64_t expirations;
                // 读取定时器到期次数，这会重置定时器内部的计数
                // 每次读取都会清除可读状态
                ssize_t s = read(timer_fd, &expirations, sizeof(expirations));
                if (s != sizeof(expirations))
                {
                    perror("read timerfd failed");
                    break;
                }

                // 处理定时器到期事件
                json_payload = serialize_info(std_array, std_num, 1,db);
                pubmsg.payload = json_payload;
                pubmsg.payloadlen = (int)strlen(json_payload);
                pubmsg.qos = QOS;
                pubmsg.retained = 0;
                MQTTClient_publishMessage(client, TOPIC, &pubmsg, &token); // 发布消息到指定主题
                free(json_payload);
            }
        }
        break;
    }
    }

    MQTTClient_unsubscribe(client, TOPIC2);
    MQTTClient_disconnect(client, 10000);
    MQTTClient_destroy(&client);
    // 关闭定时器文件描述符
    close(timer_fd);
    shm_del(&para);
    return rc;
}